package com.gzhryc.shared_device.iot.core.aliyun;

import com.aliyun.openservices.iot.api.Profile;
import com.aliyun.openservices.iot.api.message.MessageClientFactory;
import com.aliyun.openservices.iot.api.message.api.MessageClient;
import com.aliyun.openservices.iot.api.message.callback.MessageCallback;
import com.aliyun.openservices.iot.api.message.entity.Message;
import com.aliyun.openservices.iot.api.message.entity.MessageToken;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.exceptions.ServerException;
import com.aliyuncs.iot.model.v20180120.PubRequest;
import com.aliyuncs.iot.model.v20180120.PubResponse;
import com.aliyuncs.iot.model.v20180120.RRpcRequest;
import com.aliyuncs.iot.model.v20180120.RRpcResponse;
import com.aliyuncs.profile.DefaultProfile;
import com.aliyuncs.profile.IClientProfile;
import com.gzhryc.common.logging.Logger;
import com.gzhryc.shared_device.iot.core.*;
import com.gzhryc.shared_device.iot.core.aliyun.models.AliyunIOTConfig;
import com.gzhryc.shared_device.iot.core.aliyun.models.AliyunIOTSynResponse;
import com.gzhryc.shared_device.iot.core.enums.EIOTType;

import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 阿里云MQTT版物联网客户端
 * 
 * @author zhanghao
 *
 */
public class AliyunIOTClient implements IOTClient {

	static Logger log = Logger.getLogger(AliyunIOTClient.class);

	final AliyunIOTConfig mConfig;
	final IOTNodeEntity mIOTNode;
	final IOTClientEventListener mListener;

	DefaultAcsClient apiClient;
	List<MessageClient> clientList;
	Integer poolSize = 2;

	public AliyunIOTClient(AliyunIOTConfig config, IOTNodeEntity iotNode,IOTClientEventListener listener) {
		this.mConfig = config;
		this.mListener = listener;
		this.mIOTNode = iotNode;
		this.poolSize = Runtime.getRuntime().availableProcessors();
		this.clientList = new ArrayList<MessageClient>();
	}

	@Override
	public boolean send(IOTMessage msg) {
		PubRequest request = new PubRequest();
		request.setProductKey(mConfig.getProductKey());
		request.setIotInstanceId(mConfig.getIotInstanceId());
		request.setMessageContent(Base64.getEncoder().encodeToString(msg.getBytes()));
		request.setTopicFullName("/" + mConfig.getProductKey() + msg.getTopic());
		request.setQos(0); // 目前支持QoS0和QoS1
		try {
			log.info("发送命令到:" + request.getTopicFullName() + ",内容:" + msg.getString());
			PubResponse response = apiClient.getAcsResponse(request);
			if (response.getSuccess()) {
				try {
					mListener.sendMessage(msg, mConfig.getIotId(), request.getTopicFullName());
				} catch (Exception e) {
					log.error(e.getMessage(), e);
				}
				return true;
			} else {
				log.error("{{0}}节点发送失败，返回结果：{{1:json}}", mConfig.getIotId(), response);
			}
		} catch (ServerException e) {
			e.printStackTrace();
		} catch (ClientException e) {
			e.printStackTrace();
		}
		return false;
	}

	@Override
	public void run() {
		try {
			IClientProfile profile = DefaultProfile.getProfile(mConfig.getRegionId(), mConfig.getAccessKey(), mConfig.getAccessSecret());
			DefaultProfile.addEndpoint(mConfig.getRegionId(), mConfig.getRegionId(), "Iot", "iot." + mConfig.getRegionId() + ".aliyuncs.com");
			apiClient = new DefaultAcsClient(profile); // 初始化SDK客户端

			final AliyunIOTClient self = this;

			for (int i = 0; i < poolSize; i++) {
				log.info("启动第" + (i + 1) + "个接收主行程");
				new Thread() {
					public void run() {
						// 配置数据监听
						String endPoint = "https://" + mConfig.getUid() + ".iot-as-http2." + mConfig.getRegionId() + ".aliyuncs.com";
						// 连接配置
						Profile listenerProfile = Profile.getAccessKeyProfile(endPoint, mConfig.getRegionId(), mConfig.getAccessKey(),
								mConfig.getAccessSecret());
						// 构造客户端
						MessageClient client = MessageClientFactory.messageClient(listenerProfile);

						// 设置接收返回
						client.setMessageListener("/" + mConfig.getProductKey() + "/#", new MessageCallback() {
							@Override
							public Action consume(MessageToken messageToken) {
								Message msg = messageToken.getMessage();
								IOTWorkThreadFactory.get().execute(new ReceiveMessageWorkThread(self, msg));
								return MessageCallback.Action.CommitSuccess;
							}
						});

						// 数据接收
						client.connect(new MessageCallback() {
							@Override
							public Action consume(MessageToken messageToken) {
								Message msg = messageToken.getMessage();
								if (msg.getTopic().startsWith("/as/mqtt/status/")) {
									IOTWorkThreadFactory.get().execute(new StateMessageWorkThread(self, msg));
								} else {
									IOTWorkThreadFactory.get().execute(new ReceiveMessageWorkThread(self, msg));
								}
								return MessageCallback.Action.CommitSuccess;
							}
						});
						clientList.add(client);
					}
				}.start();
			}
		} catch (ClientException e) {
			e.printStackTrace();
		}
	}

	@Override
	public IOTSynResponse synSend(IOTMessage msg, Long times, TimeUnit unit) {
		RRpcRequest request = new RRpcRequest();
		request.setProductKey(mConfig.getProductKey());
		request.setIotInstanceId(mConfig.getIotInstanceId());
		request.setDeviceName(msg.getDeviceSn());
		request.setRequestBase64Byte(Base64.getEncoder().encodeToString(msg.getBytes()));
		request.setTimeout((int) unit.toMillis(times));
		// request.setTopic("/"+mConfig.getProductKey() + msg.getTopic());
		try {
			log.info("发送命令到:" + request.getTopic() + ",内容:" + msg.getString());
			RRpcResponse response = apiClient.getAcsResponse(request);
			IOTSynResponse result = new AliyunIOTSynResponse(response);
			return result;
		} catch (ServerException e) {
			e.printStackTrace();
		} catch (ClientException e) {
			e.printStackTrace();
		}
		return null;
	}

	public class ReceiveMessageWorkThread implements Runnable {

		private final AliyunIOTClient iotClient;
		private final Message msg;

		public ReceiveMessageWorkThread(AliyunIOTClient iotClient, Message msg) {
			this.iotClient = iotClient;
			this.msg = msg;
		}

		@Override
		public void run() {
			AliyunIOTClientEvent event = new AliyunIOTClientEvent(iotClient, msg.getTopic(), msg.getPayload());
			mListener.receiveMessage(event);
		}
	}

	public class StateMessageWorkThread implements Runnable {

		private final AliyunIOTClient iotClient;
		private final Message msg;

		public StateMessageWorkThread(AliyunIOTClient iotClient, Message msg) {
			this.iotClient = iotClient;
			this.msg = msg;
		}

		@Override
		public void run() {
			AliyunIOTClientEvent event = new AliyunIOTClientEvent(iotClient, msg.getTopic(), msg.getPayload());
			mListener.stateChange(event);
		}
	}

	public void unsubscribe(String topic) {
		for (MessageClient client : clientList) {
			client.unsubscribe(topic);
		}
	}

	@Override
	public void close() {
		for (MessageClient client : clientList) {
			client.unsubscribe("/" + mConfig.getProductKey() + "/#");
			client.disconnect();
		}
	}

	@Override
	public EIOTType getIOTType() {
		return EIOTType.ALIYUN;
	}

	@Override
	public IOTNodeEntity getIOTNote() {
		return mIOTNode;
	}
}
